package pub.cleangao.memm.transport.netty;


import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import io.netty.handler.timeout.IdleStateHandler;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import pub.cleangao.memm.exception.ConnectionUnavailable;
import pub.cleangao.memm.exception.RpcException;
import pub.cleangao.memm.registry.RpcRuntime;
import pub.cleangao.memm.registry.RpcServiceRegistry;
import pub.cleangao.memm.transport.RequestTransport;
import pub.cleangao.memm.transport.message.MessageType;
import pub.cleangao.memm.transport.message.RpcMessage;
import pub.cleangao.memm.transport.message.RpcRequest;
import pub.cleangao.memm.transport.message.RpcResponse;
import pub.cleangao.memm.transport.netty.codec.MessageDecoder;
import pub.cleangao.memm.transport.netty.codec.MessageEncoder;
import pub.cleangao.memm.transport.netty.compress.CompressType;
import pub.cleangao.memm.transport.netty.serialize.SerializationType;
import pub.cleangao.memm.util.ThreadPoolFactory;

import java.net.InetSocketAddress;
import java.net.Socket;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;


@Slf4j
public class NettyRpcClient implements RequestTransport {

    private final RpcServiceRegistry serviceRegistry = RpcRuntime.getRegistry();
    private final UnprocessedRequests unprocessedRequests = UnprocessedRequests.getInstance();
    ;
    private final ChannelHolder channelHolder = ChannelHolder.getInstance();

    private final Bootstrap bootstrap;
    private final EventLoopGroup eventLoopGroup;


    private static final int CPU_CORE_COUNT = Runtime.getRuntime().availableProcessors();

    private static NettyRpcClient instance;

    public static NettyRpcClient getInstance() {
        if (instance == null) {
            instance = new NettyRpcClient();
        }
        return instance;
    }

    public NettyRpcClient() {
        eventLoopGroup = new NioEventLoopGroup(CPU_CORE_COUNT * 2, ThreadPoolFactory.createThreadFactory("rpc-client", true));
        bootstrap = new Bootstrap();
        bootstrap.group(eventLoopGroup)
                .channel(NioSocketChannel.class)
                .handler(new LoggingHandler(LogLevel.INFO))
                //  The timeout period of the connection.
                //  If this time is exceeded or the connection cannot be established, the connection fails.
                .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 5000)
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel ch) {
                        ChannelPipeline p = ch.pipeline();
                        // If no data is sent to the server within 15 seconds, a heartbeat request is sent
                        p.addLast(new IdleStateHandler(0, 5, 0, TimeUnit.SECONDS));
                        p.addLast(new MessageEncoder());
                        p.addLast(new MessageDecoder());
                        p.addLast(new NettyRpcClientHandler());
                    }
                });
    }

    /**
     * connect server and get the channel ,so that you can send rpc message to server
     *
     * @param inetSocketAddress server address
     * @return the channel
     */
    @SneakyThrows
    public Channel doConnect(InetSocketAddress inetSocketAddress) {
        CompletableFuture<Channel> completableFuture = new CompletableFuture<>();
        bootstrap.connect(inetSocketAddress).addListener((ChannelFutureListener) future -> {
            if (future.isSuccess()) {
                log.info("Rpc client connected to [{}] successful.", inetSocketAddress.toString());
                completableFuture.complete(future.channel());
            } else {
                ConnectionUnavailable e = new ConnectionUnavailable("Unable to connect remote rpc server node [" + inetSocketAddress.getHostName() + ":" + inetSocketAddress.getPort() + "]");
                e.setHost(inetSocketAddress.getHostName());
                e.setPort(inetSocketAddress.getPort());
                throw e;
            }
        });
        return completableFuture.get();
    }

    @Override
    public RpcResponse<Object> send(RpcRequest rpcRequest) {
        CompletableFuture<RpcResponse<Object>> resultFuture = new CompletableFuture<>();
        doSendRequest(rpcRequest, resultFuture);
        try {
            if (rpcRequest.getTimeout() == 0L) {
                return resultFuture.get();
            } else {
                return resultFuture.get(rpcRequest.getTimeout(), TimeUnit.MILLISECONDS);
            }
        } catch (Exception e) {
            log.error("Rpc request error: " + e.getMessage());
            e.printStackTrace();
            throw new RpcException(e);
        }
    }

    private void doSendRequest(RpcRequest rpcRequest, CompletableFuture<RpcResponse<Object>> resultFuture) {
        // 1. 获取服务的地址
        InetSocketAddress inetSocketAddress = serviceRegistry.lookupService(rpcRequest);
        // 2. 获取channel
        Channel channel = getChannel(inetSocketAddress);
        if (channel.isActive()) {
            unprocessedRequests.put(rpcRequest.getRequestId(), resultFuture);
            RpcMessage rpcMessage = new RpcMessage();
            rpcMessage.setData(rpcRequest);
            rpcMessage.setCodec(SerializationType.PROTOSTUFF.getCode());
            rpcMessage.setCompress(CompressType.NONE.getCode());
            rpcMessage.setMessageType(MessageType.REQUEST.value());
            channel.writeAndFlush(rpcMessage).addListener((ChannelFutureListener) future -> {
                if (future.isSuccess()) {
                    log.debug("Client send message: [{}]", rpcMessage);
                } else {
                    future.channel().close();
                    resultFuture.completeExceptionally(future.cause());
                    log.error("Rpc message send failed: ", future.cause());
                }
            });
        } else {
            ConnectionUnavailable e = new ConnectionUnavailable();
            e.setHost(inetSocketAddress.getHostName());
            e.setPort(inetSocketAddress.getPort());
            throw e;
        }
    }


    public Channel getChannel(InetSocketAddress inetSocketAddress) {
        Channel channel = channelHolder.get(inetSocketAddress);
        if (channel == null) {
            channel = doConnect(inetSocketAddress);
            channelHolder.set(inetSocketAddress, channel);
        }
        return channel;
    }


    @Override
    public void shutdown() {
        eventLoopGroup.shutdownGracefully();
    }

    @Override
    public boolean testConnection(String host, int port) {
        InetSocketAddress inetSocketAddress = new InetSocketAddress(host, port);
        try (Socket socket = new Socket()) {
            socket.connect(inetSocketAddress);
            return true;
        } catch (Exception ignore) {
            return false;
        }
    }
}
